package bigdata;

import java.awt.event.WindowAdapter;
import java.awt.event.WindowEvent;
import java.awt.event.WindowListener;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Map;

import javax.swing.JFrame;
import javax.swing.JOptionPane;
import javax.swing.SwingUtilities;
import javax.swing.UIManager;
import javax.swing.UnsupportedLookAndFeelException;

import twitter4j.internal.org.json.JSONArray;
import twitter4j.internal.org.json.JSONException;
import twitter4j.internal.org.json.JSONObject;
import backtype.storm.Config;
import backtype.storm.LocalCluster;
import backtype.storm.StormSubmitter;
import backtype.storm.task.OutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.testing.TestWordSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.TopologyBuilder;
import backtype.storm.topology.base.BaseRichBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;
import backtype.storm.tuple.Values;
import backtype.storm.utils.Utils;

public class RunStorm {
	public static LocalCluster cluster;


	public static void run() {

		//Start mysql
		SimulationMain.gui.addStringToLog("Starting mysql...");
		SqlDB.runDB();
		SimulationMain.gui.addStringToLog("mysql started.");
		SqlDB.cleanCustomersTable();
		Utils.sleep(1000);
		SimulationMain.gui.addStringToLog("Starting storm...");

		TopologyBuilder builder = new TopologyBuilder();

		builder.setSpout("line", new InputSpout(), 1);
		builder.setBolt("FilterBolt", new FilterBolt(), 1).shuffleGrouping(
				"line");
		builder.setBolt("updateBolt", new UpdateBolt(), 1).shuffleGrouping(
				"FilterBolt");
		// builder.setBolt("exclaim2", new ExclamationBolt(),
		// 2).shuffleGrouping("exclaim1");

		Config conf = new Config();
		conf.setDebug(true);
		
		cluster = new LocalCluster();

		cluster.submitTopology("test", conf, builder.createTopology());

		SimulationMain.gui.addStringToLog("Storm is running.");

		(new Thread(new UpdateAvergeThread())).start();
		Utils.sleep(120000);

		// cluster.killTopology("test");
		// cluster.shutdown();
	}

}
